import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

// MapReduce任务的Driver程序
public class WeiBoUserCountDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // 创建hadoop conf
        Configuration conf = new Configuration();
        // 配置HDFS URL
        conf.set("fs.defaultFS", "hdfs://master:9000");
        // 设置副本的数量
        conf.set("dfs.replication", "1");
        // 通过conf创建job
        Job job = Job.getInstance(conf);
        // 设置job名字
        job.setJobName("WeiBoUserCount");
        // 配置MapReduce任务运行的入口程序
        job.setJarByClass(WeiBoUserCountDriver.class);

        // 配置Map任务
        // 配置Map任务执行的class
        job.setMapperClass(WeiBoUserCountMap.class);
        // 配置Map任务输出的key的类型
        job.setMapOutputKeyClass(Text.class);
        // 配置Map任务输出的value的类型
        job.setMapOutputValueClass(IntWritable.class);

        // 配置Reduce任务
        // 配置Reduce任务执行的class
        job.setReducerClass(WeiBoUserCountReduce.class);
        // 配置Reduce任务输出的key的类型
        job.setOutputKeyClass(Text.class);
        // 配置Reduce任务输出的value的类型
        job.setOutputValueClass(IntWritable.class);


        // 配置输入路径
        FileInputFormat.addInputPath(job, new Path("/WeiBoUser/input"));

        // 构建FileSystem对象
        FileSystem fs = FileSystem.get(conf);
        // 判断输出路径是否存在 存在就将其删除 否则每次重新运行MR任务时需要手动删除
        if (fs.exists(new Path("/WeiBoUser/output"))) {
            fs.delete(new Path("/WeiBoUser/output"), true);
        }
        // 配置输出路径
        FileOutputFormat.setOutputPath(job, new Path("/WeiBoUser/output"));
        // 等待任务运行完成
        job.waitForCompletion(true);

    }
    /**
     *  1、在HDFS中创建/WeiBoUser/input目录
     *  hdfs dfs -mkdir -p /WeiBoUser/input
     *  2、将user_clean.csv文件上传至虚拟机再通过HDFS Shell命令上传至HDFS的/WeiBoUser/input目录中
     *  hdfs dfs -put user_clean.csv /WeiBoUser/input
     *  3、使用maven将项目打成jar包并上传至虚拟机，然后启动MapReduce任务
     *  hadoop jar WeiBoUserCount-1.0.jar WeiBoUserCountDriver
     *  4、查看运行结果
     *  hdfs dfs -cat /WeiBoUser/output/part-r-00000
     */
}
